package com.netty.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.topsoft.common.AppContext;
import com.topsoft.common.BaseMessage;
import com.topsoft.common.MiscHelper;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.data.redis.core.RedisTemplate;

import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public final class WebSocketClient implements Runnable{
    private static int overLengthIsCompress=10;//消息超过多长就压缩
    private EventLoopGroup loop = new NioEventLoopGroup();
    private Class<? extends WebSocketClientHandler> clazzWebSocketClientHandler;
    private String url;
    private URI __URI;
    public QueueType queueType;
    private String queue;
    private JSONObject parameters;
    private long reconnectInterval=3;

    //redis缓存消息失败, 则缓存到本地
    private static ConcurrentHashMap<String,BaseMessage> mapBaseMessage =new ConcurrentHashMap();

    public Channel channel;
    public String keyName;
    public static ConcurrentHashMap<String,WebSocketClient> mapWebSocketClient =new ConcurrentHashMap();
    public static ConcurrentHashMap<String,Channel> queuechannels =new ConcurrentHashMap();
    public static Logger logger = LogManager.getLogger("【websocket.client】");


    public static WebSocketClient urlSystemTest=null;
    public static void main( String[] args ) {
        //测试
        urlSystemTest= new WebSocketClient(SystemHandler.class,"websocket.urlSystem");
//        WebSocketClient WebSocketClient= new WebSocketClient(SystemHandler.class,"websocket.urlTiulo2Erp");
        new Thread(urlSystemTest).start();//开启新线程
//        new Thread(WebSocketClient).start();//开启新线程
    }

    public WebSocketClient(Class<? extends WebSocketClientHandler> clazz,String keyName){
        this.clazzWebSocketClientHandler=clazz;
        //String temp= null;//dbcProperties.getString(keyName);
        String temp= AppContext.environment().getProperty(keyName);
        if (temp==null || temp.trim().length()==0){
            this.url=keyName.trim();
            this.keyName = null;
        }else {
            this.url=temp.trim();
            this.keyName = keyName;
            mapWebSocketClient.put(this.keyName, this);
        }

        try {
            if (url.indexOf("&queueType")<0) url=url+"&queueType=P_ONE";

            __URI = new URI(url);
            parameters= MiscHelper.urlParameter2Map(__URI.getQuery());
            String queueType2 =parameters.getString("queueType");
            if (!queueType2.equals(QueueType.PC_ONE.name()) && !queueType2.equals(QueueType.PC_MORE.name())
                    && !queueType2.equals(QueueType.C_ONE.name()) && !queueType2.equals(QueueType.C_MORE.name())
                    && !queueType2.equals(QueueType.P_ONE.name()) && !queueType2.equals(QueueType.P_MORE.name())){
                throw new RuntimeException("queueType参数不对，正确值为："+QueueType.values().toString());
            }

            queueType=QueueType.valueOf(queueType2);
            queue=parameters.getString("queue");
        } catch (URISyntaxException e) {
            e.printStackTrace();
        }
    }

    public void run() {
        createBootstrap(new Bootstrap(), loop);
    }

    public String getUrl(){
        return this.url;
    }
    public URI getURL(){
        return this.__URI;
    }

    public Bootstrap createBootstrap(Bootstrap bootstrap, EventLoopGroup eventLoop) {
        if (url==null && url.length()<2 || url.indexOf("queue=")<0) {
            logger.error("连接终止，地址："+url+"  必须有队列参数queue。");
            return null;
        }
        if (bootstrap != null) {
            final WebSocketClientHandler handler;
            try {
                handler =clazzWebSocketClientHandler.newInstance();
                WebSocketClientHandshaker webSocketClientHandshaker=WebSocketClientHandshakerFactory.newHandshaker(__URI,WebSocketVersion.V13, null, false, new DefaultHttpHeaders());
                handler.init(this,webSocketClientHandshaker);

            } catch (Exception e) {
                logger.error("尝试连接 "+url+"  出现错误"+e.getMessage());
                return null;
            }

            bootstrap.group(eventLoop);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);//
            bootstrap.channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {
                @Override protected void initChannel(SocketChannel ch) {
                    ChannelPipeline channelPipeline = ch.pipeline();
                    channelPipeline.addLast(new HttpClientCodec(), new HttpObjectAggregator(8192), handler);
                }
            });
            ChannelFuture channelFuture=bootstrap.connect(__URI.getHost(), __URI.getPort()).addListener(new WebSocketConnection(this));

            try {
                if(MiscHelper.isNull(queue)){
                    logger.error("连接终止，地址："+url+"  必须有队列参数queue。");
                }else{
                    Channel channel =channelFuture.sync().channel();
//                    channels.put(queue,channel);
                    handler.handshakeFuture().sync();
                }
            } catch (InterruptedException e1) {
                logger.error("尝试连接 "+url+"  出现错误");
                e1.printStackTrace();
            } catch (Exception e2) {
                logger.error("尝试连接 "+url+" 出现错误");
            }finally{

            }
            ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
        }
        return bootstrap;
    }

    //static RedisTemplate<String, BaseMessage> redisTemplate;
//
//    public static boolean sendMessage(Map<String, String> mapMessage) {
//        List<BaseMessage> list = new ArrayList<>(1);
//        list.add(baseMessage);
//        return sendMessage(recvQueue, list);
//    }
    public static boolean sendMessage(String recvQueue, List<String> listMessage){
        try {
            WebSocketClient webSocketClientBySystem=mapWebSocketClient.get("websocket.urlSystem");
            WebSocketClient webSocketClient=null;
            Channel channelSend=null;
            if (recvQueue!=null && recvQueue.trim().length()>0){
                webSocketClient=mapWebSocketClient.get(recvQueue.trim());
                if (webSocketClient==null) {
                    logger.info( " 找不到keyName="+recvQueue+"的WebSocketClient,将以系统队列发送消息");
                }else{
                    channelSend = webSocketClient.channel;
                }
            }
            if (webSocketClient==null && webSocketClientBySystem==null) {
                logger.error(recvQueue + " 找不到可用于发送消息的channel对象");
                return false;
            }
            if (webSocketClient==null && webSocketClientBySystem!=null) {
                channelSend = webSocketClientBySystem.channel;
            }

            if (channelSend==null) {
                logger.error(recvQueue +" webSocketClient.channel 为null");
                return false;
            }
            if (channelSend.isOpen()==false) {
                logger.error(recvQueue + " 的webSocketClient.channel的连接已经关闭");
                return false;
            }

            StringBuilder sbData = new StringBuilder();
//            Object data=baseMessage.data();
//            if(data instanceof Map){
//                Map mapData = (Map)data;
//                sbData.append(JSON.toJSONString(mapData));
//            }else if(data instanceof String){
//                sbData.append(data);
//            }
//
//            if (data instanceof List){
//                List listData = (List)data;
//
//
//            }


            String msgString = listMessage.toString();
            if(msgString.length()>overLengthIsCompress){
                msgString="__gzip:"+ MiscHelper.gzip2String(msgString,"UTF-8");
                channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
                logger.error("发送消息:{}",msgString);

            }else {
                channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
            }

            sbData.setLength(0); sbData=null; msgString=null;
        } catch (Exception e2) {
            logger.error("WebSocketClient.sendMessage 报错："+e2.getMessage());
        }

        return true;
    }

//2020-01-19
//    public static void sendMessage(String keyName,String __recvQueue,JSONObject msgJson){
//        keyName=keyName.trim();
//        if (__recvQueue==null || __recvQueue.trim().equals("")){
//            msgJson.remove("__queue");
//        }else {
//            msgJson.put("__queue", __recvQueue.trim());
//        }
//
//        try {
//            WebSocketClient webSocketClientBySystem=mapWebSocketClient.get("websocket.urlSystem");
//            WebSocketClient webSocketClient=null;
//            Channel channelSend=null;
//            if (keyName!=null && keyName.trim().length()>0){
//                webSocketClient=mapWebSocketClient.get(keyName.trim());
//                if (webSocketClient==null) {
//                    logger.error( " 找不到keyName="+keyName+"的WebSocketClient,将以系统队列发送消息");
//                }else{
//                    channelSend = webSocketClient.channel;
//                }
//            }
//            if (webSocketClient==null && webSocketClientBySystem==null) {
//                logger.error(keyName + " 找不到可用于发送消息的channel对象");
//                return;
//            }
//            if (webSocketClient==null && webSocketClientBySystem!=null) {
//                channelSend = webSocketClientBySystem.channel;
//            }
//
//            if (channelSend==null) {
//                logger.error(keyName +" webSocketClient.channel 为null");
//                return;
//            }
//            if (channelSend.isOpen()==false) {
//                logger.error(keyName + " 的webSocketClient.channel的连接已经关闭");
//                return;
//            }
//
//            String msgString=msgJson.toJSONString();
//            if(msgString.length()>overLengthIsCompress){
//                msgString="__gzip:"+ MiscHelper.gzip2String(msgString,"UTF-8");
//                channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
//            }else {
//                channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
//            }
//
//            msgJson.clear();msgJson=null;
//            msgString=null;
//        } catch (Exception e2) {
//            logger.error("WebSocketClient.sendMessage 报错："+e2.getMessage());
//        }
//    }

//    public static void sendMessage(String __sendQueue,String __recvQueue,JSONObject msgJson){
//        if (__recvQueue==null || __recvQueue.equals("")){
//            logger.warning("接收队列名不能为空");
//        }
//        if (Misc.isNull(__recvQueue)==false) msgJson.put("__queue",__recvQueue);
//
//        try {
//            Channel channelSend = queuechannels.get(__sendQueue);
//            if (channelSend==null) {
//                logger.warning("发送的队列" + __sendQueue + "未启动");
//            }else if (channelSend.isOpen()==false){
//                logger.warning("发送的队列"+__sendQueue+"的连接已经关闭");
//            } else {
//                String msgString=msgJson.toJSONString();
//                if(msgString.length()>overLengthIsCompress){
//                    msgString="__gzip:"+MiscHelper.gzip2String(msgString,"UTF-8");
//                    //msgString=URLEncoder.encode(msgString,"UTF-8");//转码，解决前段 解压后 中文的问题
////                    msgString=MiscHelper.gzip2(msgString);
//                    //channelSend.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(MessageGZIP.compressToByte(msgString))));
//                    channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
//                    //channelSend.writeAndFlush(new BinaryWebSocketFrame(Unpooled.wrappedBuffer(MiscHelper.gzip2byte(msgString,"UTF-8"))));
//                }else {
//                    channelSend.writeAndFlush(new TextWebSocketFrame(msgString));
//                }
//            }
//        } catch (Exception e2) {
//            logger.warning("系统队列system报错："+e2.getMessage());
//        }
//    }

    /**
     * 获得队列名称
     * @return
     */
    public String getQueue(){
        return this.queue;
    }
    /**
     * 获得连接参数
     * @return
     */
    public JSONObject getParameters(){
        return this.parameters;
    }


    /**
     * 获得试图重新连接之前延迟的秒数。默认值：3（秒）
     * @return
     */
    public long getReConnectInterval(){
        return this.reconnectInterval;
    }
    /**
     * 设置试图重新连接之前延迟的秒数。默认值：3（秒）
     * @return
     */
    public void setReConnectInterval(long second){
        this.reconnectInterval=second;
    }
}
